package org.apache.kylin.common.streaming;

import com.google.common.base.Preconditions;
import com.google.common.collect.Maps;
import org.apache.kylin.common.util.Pair;

import java.io.Serializable;
import java.util.Iterator;
import java.util.Map;

/**
 * Created by taoliu on 2017/2/6.
 */
public class BatchDesc implements Serializable,Comparable<BatchDesc>{
    private long id = System.currentTimeMillis();
    private String cubeName;
    private Pair<Long, Long> timeRange;
    private Map<Integer,Pair<Long,Long>> partitionOffsetRanges = Maps.newHashMap();
    private boolean isGap = false;
    private boolean isValid = true;


    public BatchDesc(String cubeName, Pair<Long, Long> timeRange,boolean isGap, Map<Integer, Pair<Long, Long>> partitionOffsetRanges) {
        this.cubeName = Preconditions.checkNotNull(cubeName);
        Preconditions.checkArgument(timeRange !=null && timeRange.getFirst() < timeRange.getSecond());
        this.timeRange = timeRange;
        this.isGap = isGap;
        this.partitionOffsetRanges = partitionOffsetRanges;
        boolean timeRangeValid = timeRange.getFirst() < timeRange.getSecond();
        if (isGap){
            Preconditions.checkNotNull(this.partitionOffsetRanges);
            Iterator<Pair<Long, Long>> iterator = this.partitionOffsetRanges.values().iterator();
            boolean pairValid = false;
            while (iterator.hasNext()){
                Pair<Long, Long> pair = iterator.next();
                if (pair.getSecond() - pair.getFirst() >= 2){
                    pairValid = true;
                    break;
                }
            }
            this.isValid = timeRangeValid && pairValid;
        }else {
            this.isValid = timeRangeValid;
        }

    }

    public long getId() {
        return id;
    }

    public boolean isValid() {
        return isValid;
    }

    public boolean isGap(){
        return isGap;
    }

    public Pair<Long, Long> getTimeRange() {
        return timeRange;
    }

    public Map<Integer, Pair<Long, Long>> getPartitionOffsetRanges() {
        return partitionOffsetRanges;
    }

    public String getCubeName() {
        return cubeName;
    }

    public void setCubeName(String cubeName) {
        this.cubeName = cubeName;
    }

    public void setTimeRange(Pair<Long, Long> timeRange) {
        this.timeRange = timeRange;
    }

    public void setPartitionOffsetRanges(Map<Integer, Pair<Long, Long>> partitionOffsetRanges) {
        this.partitionOffsetRanges = partitionOffsetRanges;
    }

    @Override
    public String toString() {
        return "BatchDesc{" +
                "id=" + id +
                ", cubeName='" + cubeName + '\'' +
                ", timeRange=" + timeRange +
                ", partitionOffsetRanges=" + partitionOffsetRanges +
                ", isGap=" + isGap +
                '}';
    }

    @Override
    public int compareTo(BatchDesc o) {
        Pair<Long, Long> timeRange = this.getTimeRange();
        Pair<Long, Long> other = o.getTimeRange();
        long comp = timeRange.getFirst() - other.getFirst();
        if (comp != 0)
            return comp < 0 ? -1 : 1;

        comp = timeRange.getSecond() - other.getSecond();
        if (comp != 0)
            return comp < 0 ? -1 : 1;
        else
            return 0;
    }
}
